{#
 Licensed to the Apache Software Foundation (ASF) under one
 or more contributor license agreements.  See the NOTICE file
 distributed with this work for additional information
 regarding copyright ownership.  The ASF licenses this file
 to you under the Apache License, Version 2.0 (the
 "License"); you may not use this file except in compliance
 with the License.  You may obtain a copy of the License at

   http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing,
 software distributed under the License is distributed on an
 "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 KIND, either express or implied.  See the License for the
 specific language governing permissions and limitations
 under the License.
#}

{#
 Licensed to the Apache Software Foundation (ASF) under one
 or more contributor license agreements.  See the NOTICE file
 distributed with this work for additional information
 regarding copyright ownership.  The ASF licenses this file
 to you under the Apache License, Version 2.0 (the
 "License"); you may not use this file except in compliance
 with the License.  You may obtain a copy of the License at

   http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing,
 software distributed under the License is distributed on an
 "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 KIND, either express or implied.  See the License for the
 specific language governing permissions and limitations
 under the License.
#}

{% extends base_template %}
{% from 'airflow/asset_next_run_modal.html' import asset_next_run_modal %}
{% from 'appbuilder/dag_docs.html' import dag_docs %}

{% block page_title %}{{ dag.dag_display_name }} - {{ appbuilder.app_name }}{% endblock %}

{% block head_css %}
  {{ super() }}
  <link rel="stylesheet" type="text/css" href="{{ url_for_asset('switch.css') }}">
{% endblock %}

{% set dag_is_paused = dag.get_is_paused() %}
{% set base_date_arg = request.args.get('base_date') %}
{% set num_runs_arg = request.args.get('num_runs') %}
{% if logical_date is defined %}
  {% set execution_date_arg = logical_date %}
{% else %}
  {% set execution_date_arg = request.args.get('logical_date') %}
{% endif %}

{% block head_meta %}
  {{ super() }}
  <meta name="dag_id" content="{{ dag.dag_id }}">
  <meta name="dag_display_name" content="{{ dag.dag_display_name }}">
  <meta name="dag_timezone" content="{{ dag.timezone.name }}">
  <!-- Urls -->
  <meta name="logs_with_metadata_url" content="{{ url_for('Airflow.get_logs_with_metadata') }}">
  <meta name="external_log_url" content="{{ url_for('Airflow.redirect_to_external_log') }}">
  <meta name="extra_links_url" content="{{ url_for('Airflow.extra_links') }}">
  <meta name="paused_url" content="{{ url_for('Airflow.paused') }}">
  <meta name="dagrun_clear_url" content="{{ url_for('Airflow.dagrun_clear') }}">
  <meta name="dagrun_failed_url" content="{{ url_for('Airflow.dagrun_failed') }}">
  <meta name="dagrun_success_url" content="{{ url_for('Airflow.dagrun_success') }}">
  <meta name="dagrun_queued_url" content="{{ url_for('Airflow.dagrun_queued') }}">
  <meta name="clear_url" content="{{ url_for('Airflow.clear') }}">
  <meta name="failed_url" content="{{ url_for('Airflow.failed') }}">
  <meta name="success_url" content="{{ url_for('Airflow.success') }}">
  <meta name="confirm_url" content="{{ url_for('Airflow.confirm') }}">
  <meta name="grid_data_url" content="{{ url_for('Airflow.grid_data') }}">
  <meta name="graph_data_url" content="{{ url_for('Airflow.graph_data') }}">
  <meta name="calendar_data_url" content="{{ url_for('Airflow.calendar_data') }}">
  <meta name="next_run_assets_url" content="{{ url_for('Airflow.next_run_assets', dag_id=dag.dag_id) }}">
  <meta name="grid_url" content="{{ url_for('Airflow.grid', dag_id=dag.dag_id) }}">
  <meta name="task_tries_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_task_instance_endpoint_get_task_instance_tries', dag_id='_DAG_ID_', dag_run_id='_DAG_RUN_ID_', task_id='_TASK_ID_') }}">
  <meta name="assets_url" content="{{ url_for('Airflow.assets') }}">
  <meta name="grid_url_no_root" content="{{ url_for('Airflow.grid', dag_id=dag.dag_id, num_runs=num_runs_arg, base_date=base_date_arg) }}">
  <meta name="graph_url" content="{{ url_for('Airflow.graph', dag_id=dag.dag_id, root=root) }}">
  <meta name="task_url" content="{{ url_for('Airflow.task', dag_id=dag.dag_id) }}">
  <meta name="log_url" content="{{ url_for('Airflow.log', dag_id=dag.dag_id) }}">
  <meta name="rendered_k8s_data_url" content="{{ url_for('Airflow.rendered_k8s_data', dag_id=dag.dag_id) }}">
  <meta name="task_instances_list_url" content="{{ url_for('TaskInstanceModelView.list') }}">
  <meta name="tag_index_url" content="{{ url_for('Airflow.index', tags='_TAG_NAME_') }}">
  <meta name="mapped_instances_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_task_instance_endpoint_get_mapped_task_instances', dag_id=dag.dag_id, dag_run_id='_DAG_RUN_ID_', task_id='_TASK_ID_') }}">
  <meta name="task_log_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_log_endpoint_get_log', dag_id=dag.dag_id, dag_run_id='_DAG_RUN_ID_', task_id='_TASK_ID_', task_try_number='-1') }}">
  <meta name="task_xcom_entries_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_xcom_endpoint_get_xcom_entries', dag_id=dag.dag_id, dag_run_id='_DAG_RUN_ID_', task_id='_TASK_ID_') }}">
  <meta name="task_xcom_entry_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_xcom_endpoint_get_xcom_entry', dag_id=dag.dag_id, dag_run_id='_DAG_RUN_ID_', task_id='_TASK_ID_', xcom_key='_XCOM_KEY_') }}">
  <meta name="upstream_asset_events_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_dag_run_endpoint_get_upstream_asset_events', dag_id=dag.dag_id, dag_run_id='_DAG_RUN_ID_') }}">
  <meta name="task_instance_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_task_instance_endpoint_get_task_instance', dag_id='_DAG_ID_', dag_run_id='_DAG_RUN_ID_', task_id='_TASK_ID_') }}">
  <meta name="set_task_instance_note" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_task_instance_endpoint_set_task_instance_note', dag_id=dag.dag_id, dag_run_id='_DAG_RUN_ID_', task_id='_TASK_ID_' ) }}">
  <meta name="set_mapped_task_instance_note" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_task_instance_endpoint_set_mapped_task_instance_note', dag_id=dag.dag_id, dag_run_id='_DAG_RUN_ID_', task_id='_TASK_ID_', map_index=0 ) }}">
  <meta name="set_dag_run_note" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_dag_run_endpoint_set_dag_run_note', dag_id=dag.dag_id, dag_run_id='_DAG_RUN_ID_') }}">
  <meta name="dag_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_dag_endpoint_get_dag', dag_id=dag.dag_id) }}">
  <meta name="dag_source_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_dag_source_endpoint_get_dag_source', dag_id=dag.dag_id) }}">
  <meta name="dag_details_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_dag_endpoint_get_dag_details', dag_id=dag.dag_id) }}">
  <meta name="assets_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_asset_endpoint_get_assets') }}">
  <meta name="event_logs_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_event_log_endpoint_get_event_logs') }}">
  <meta name="audit_log_url" content="{{ url_for('LogModelView.list') }}">
  <meta name="dag_run_url" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_dag_run_endpoint_get_dag_run', dag_id='__DAG_ID__', dag_run_id='__DAG_RUN_ID__') }}">
  <meta name="task_dependency_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_task_instance_endpoint_get_task_instance_dependencies', dag_id=dag.dag_id, dag_run_id='_DAG_RUN_ID_', task_id='_TASK_ID_') }}">
  <meta name="mapped_task_dependency_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_task_instance_endpoint_get_mapped_task_instance_dependencies', dag_id=dag.dag_id, dag_run_id='_DAG_RUN_ID_', task_id='_TASK_ID_', map_index=0) }}">
  <meta name="task_detail_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_task_endpoint_get_task', dag_id=dag.dag_id, task_id='_TASK_ID_') }}">

  <!-- End Urls -->
  <meta name="is_paused" content="{{ dag_is_paused }}">
  <meta name="csrf_token" content="{{ csrf_token() }}">
  <meta name="k8s_or_k8scelery_executor" content="{{ k8s_or_k8scelery_executor }}">
  <meta name="excluded_audit_log_events" content="{{ excluded_events_raw }}">
  <meta name="included_audit_log_events" content="{{ included_events_raw }}">
  {% if dag_model is defined and dag_model.next_dagrun_create_after is defined and dag_model.next_dagrun_create_after is not none %}
    <meta name="next_dagrun_create_after" content="{{ dag_model.next_dagrun_create_after }}">
    <meta name="next_dagrun_data_interval_start" content="{{ dag_model.next_dagrun_data_interval_start }}">
    <meta name="next_dagrun_data_interval_end" content="{{ dag_model.next_dagrun_data_interval_end }}">
  {% endif %}
  {% if show_external_log_redirect is defined %}
    <meta name="show_external_log_redirect" content="{{ show_external_log_redirect }}">
  {% endif %}
  {% if external_log_name is defined %}
  <meta name="external_log_name" content="{{ external_log_name }}">
  {% endif %}
  <meta name="can_edit" content="{{ dag.can_edit }}">
{% endblock %}

{% block content %}
  <div style="display: flex; align-items: center; justify-content: space-between;">
    <div>
      <h3>
        {% if can_edit_dag %}
          {% set switch_tooltip = 'Pause/Unpause DAG' %}
        {% else %}
          {% set switch_tooltip = 'DAG is Paused' if dag_is_paused else 'DAG is Active' %}
        {% endif %}
        <label class="switch-label{{' disabled' if not can_edit_dag else ''  }} js-tooltip" title="{{ switch_tooltip }}">
          <input class="switch-input" id="pause_resume" data-dag-id="{{ dag.dag_id }}"
                  {{ "data-require-confirmation" if appbuilder.require_confirmation_dag_change else "" }}
                  type="checkbox"{{ " checked" if not dag_is_paused else "" }}
                  {{ " disabled" if not can_edit_dag else "" }}>
          <span class="switch" aria-hidden="true"></span>
        </label>
        <span class="text-muted">DAG:</span> <span title="{{ dag.dag_id }}">{{ dag.dag_display_name }}</span>
        <small class="text-muted">{{ dag.description[0:150] + '…' if dag.description and dag.description|length > 150 else dag.description|default('', true) }}</small>
        {% if dag_model is defined and dag_model.max_consecutive_failed_dag_runs is defined and  dag_model.max_consecutive_failed_dag_runs > 0  %}
          <span class="material-icons text-muted js-tooltip" aria-hidden="true" data-original-title="Auto Pauses: After {{ dag_model.max_consecutive_failed_dag_runs|string }} consecutive failures">motion_photos_paused</span>
        {% endif %}
      </h3>
    </div>
    <div>
      <h4 class="js-asset-triggered" style="user-select: none;-moz-user-select: auto; display: inline-block;">
        {% if state_token is defined and state_token %}
          {{ state_token }}
        {% endif %}
        <a class="label label-default" href="{{ url_for('DagRunModelView.list') }}?_flt_3_dag_id={{ dag.dag_id }}">
          Schedule: {{ dag_model is defined and dag_model and dag_model.timetable_summary }}
        </a>
        {% if dag_model is defined and dag_model and dag_model.timetable_description %}
            <span class="material-icons text-muted js-tooltip" aria-hidden="true" data-original-title="Schedule: {{ dag_model.timetable_description|string }}">info</span>
        {% endif %}
        {% if dag_model is defined and dag_model.next_dagrun is defined and dag_model.asset_expression != None %}
          <p class="label label-default js-tooltip" style="margin-left: 5px" id="next-run" data-html="true" data-placement="bottom">
            Next Run ID: <time datetime="{{ dag_model.next_dagrun }}" data-with-tz="true">{{ dag_model.next_dagrun }}</time>
          </p>
        {% endif %}
        {% if dag_model is defined and dag_model.asset_expression is defined and dag_model.asset_expression != None %}
          {%- with asset_info = dag_model.get_asset_triggered_next_run_info() -%}
            <span
              id="next-asset-tooltip"
              class="js-tooltip"
              title="Click to see asset details."
              data-html="true"
              data-placement="bottom"
              data-uri="{{ asset_info.uri }}"
            >
                <p
                  class="label label-default next-asset-triggered"
                  style="margin-left: 5px;"
                  data-summary="
                    {%- if asset_info.total == 1 -%}
                    On {{ asset_info.uri }}
                    {%- else -%}
                    {{ asset_info.ready }} of {{ asset_info.total }} assets updated
                    {%- endif -%}"
                  >
                  {% if asset_info.total == 1 -%}
                    On {{ asset_info.uri }}
                  {%- else -%}
                  {{ asset_info.ready }} of {{ asset_info.total }} assets updated
                  {%- endif %}
                </p>
            </span>
          {%- endwith -%}
        {% endif %}
      </h4>
      <div class="btn-group">
        {% if show_trigger_form_if_no_params %}
        <div class="dropdown">
          <a aria-label="Trigger DAG" class="btn btn-default btn-icon-only{{ ' disabled' if not dag.can_trigger }} trigger-dropdown-btn" data-toggle="dropdown">
            <span class="material-icons" aria-hidden="true">play_arrow</span>
            </a>
          <ul class="dropdown-menu trigger-dropdown-menu">
            <li>
              <form method="POST" action="{{ url_for('Airflow.trigger', dag_id=dag.dag_id) }}">
                <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
                <input type="hidden" name="dag_id" value="{{ dag.dag_id }}">
                <input type="hidden" name="unpause" value="True">
                <input type="hidden" name="conf" value="{}">
                <!-- for task instance detail pages, dag_id is still a query param -->
                {% if 'dag_id' in request.args %}
                  <input type="hidden" name="origin" value="{{ url_for(request.endpoint, **request.args) }}">
                {% else %}
                  <input type="hidden" name="origin" value="{{ url_for(request.endpoint, dag_id=dag.dag_id, **request.args) }}">
                {% endif %}
                <button type="submit" class="dropdown-form-btn">Trigger DAG</button>
              </form>
            </li>
            <li>
              {% if 'dag_id' in request.args %}
                <a href="{{ url_for('Airflow.trigger', dag_id=dag.dag_id, origin=url_for(request.endpoint, **request.args)) }}">
              {% else %}
                <a href="{{ url_for('Airflow.trigger', dag_id=dag.dag_id, origin=url_for(request.endpoint, dag_id=dag.dag_id, **request.args)) }}">
              {% endif %}
              Trigger DAG w/ config</a></li>
          </ul>
        </div>
        {% else %}
        {% if 'dag_id' in request.args %}
          <a href="{{ url_for('Airflow.trigger', dag_id=dag.dag_id, origin=url_for(request.endpoint, **request.args)) }}"
        {% else %}
          <a href="{{ url_for('Airflow.trigger', dag_id=dag.dag_id, origin=url_for(request.endpoint, dag_id=dag.dag_id, **request.args)) }}"
        {% endif %}
            onclick="return {{ 'triggerDag' if not appbuilder.require_confirmation_dag_change else 'confirmTriggerDag' }}(this, '{{ dag.dag_id }}')"
            title="Trigger&nbsp;DAG"
            aria-label="Trigger DAG"
            class="btn btn-default btn-icon-only{{ ' disabled' if not dag.can_trigger }} trigger-dropdown-btn">
            <span class="material-icons" aria-hidden="true">play_arrow</span>
        </a>
        {% endif %}
        <a href="{{ url_for('Airflow.parse_dag', file_token=dag_file_token, redirect_url=url_for(request.endpoint, dag_id=dag.dag_id)) }}"
           title="Reparse&nbsp;DAG"
           aria-label="Reparse DAG"
           class="btn btn-default btn-icon-only {{ ' disabled' if not dag.can_edit }}">
          <span class="material-icons">restore_page</span>
        </a>
        <a href="{{ url_for('Airflow.delete', dag_id=dag.dag_id, redirect_url=url_for(request.endpoint, dag_id=dag.dag_id)) }}"
          title="Delete&nbsp;DAG"
          aria-label="Delete DAG"
          class="btn btn-default btn-icon-only{{ ' disabled' if not dag.can_delete }}"
          onclick="return confirmDeleteDag(this, '{{ dag.safe_dag_id }}')">
          <span class="material-icons text-danger" aria-hidden="true">delete_outline</span>
        </a>
      </div>
    </div>
  </div>
  <div class="clearfix"></div>
  {{ dag_docs(doc_md) }}
  <!-- Modal for asset-triggered next run -->
  {{ asset_next_run_modal(id='asset-next-run-modal') }}
{% endblock %}
{% block tail %}
  {{ super() }}
  <script src="{{ url_for_asset('dag.js') }}"></script>
  <script>
    // Tests rely on confirmDeleteDag to be in the html
    function confirmDeleteDag(link, id) {
      if (confirm(`Are you sure you want to delete '${id}' now?\n\
        This option will delete ALL metadata, DAG runs, etc.\n\
        EXCEPT Log.\n\
        This cannot be undone.`)) {
        postAsForm(link.href, {});
      }
      return false;
    }

    function triggerDag(link, dagId) {
      postAsForm(link.href, {});
      return false;
    }

    function confirmTriggerDag(link, dagId) {
      if (confirm(`Are you sure you want to trigger '${dagId}' now?`)) {
        postAsForm(link.href, {});
      }
      return false;
    }

  </script>
{% endblock %}
